import CATALOGMANAGER.CatalogManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.io.*;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Enumeration;

/**
 * Region_Server's zk service
 */
@Slf4j
public class ZookeeperManager {
    /**
     * zk service's ip-port
     */
    static final String ZOOKEEPER_SERVER = "localhost:2181";
    /**
     * curator holder
     */
    static CuratorFramework client;
    /**
     * Region_Server's list of table names
     */
    static ArrayList<String> table_list;
    /**
     * Region_Server's ip
     */
    static String REGION_SERVER_IP;
    static String path = "/";
    /**
     * Master_Server's ip and port
     * TODO: change master server's ip when deploying; add table_catalog and index_catalog
     *
     */
    static final String MASTER_SERVER_IP = "localhost";
    static final int MASTER_PORT = 8002;

    /**
     * @author cz
     * @date 2022/5/18
     * connect to zk and create region server's node in zk
     */
    public static void zookeeperConnect() {
        //use MINISQL's API to get the table name list
        table_list = CatalogManager.get_tables();
        System.out.println("(init) table_list: " + table_list);

        try {
            //get host ip
            REGION_SERVER_IP = String.valueOf(getLocalHostLANAddress()).substring(1);
        } catch (UnknownHostException e) {
            log.warn("cannot get region's ip");
            e.printStackTrace();
        }

        //connect with master and send region's own ip
        //then get region's path name from master
        path = getRegionName();

        //connect with zk
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 5);
        client = CuratorFrameworkFactory.builder()
                .connectString(ZOOKEEPER_SERVER)
                .retryPolicy(retryPolicy)
                .namespace("service")
                .build();
        client.start();

        //construct data: (region_ip table_name@1 table_name@2 ...)
        StringBuilder data = new StringBuilder(REGION_SERVER_IP);
        for (String t : table_list) {
            data.append(" ");
            data.append(t);
        }

        try {
            Stat isExist = client.checkExists().forPath(path);
            if (isExist != null) {
                client.delete().forPath(path);
            } else {
                //use ftp to store all tables and index files for backup
                FTPConnector.uploadFile("table_catalog");
                FTPConnector.uploadFile("index_catalog");
                for (String t : table_list) {
                    FTPConnector.uploadFile(t);
                    FTPConnector.uploadFile(t + "_index.index");
                }
            }
            //create zk node which value is the region's ip and table list
            client.create().withMode(CreateMode.EPHEMERAL).forPath(path, data.toString().getBytes());
        } catch (Exception e) {
            log.warn("ftp utils failed!");
            e.printStackTrace();
        }
    }

    public static String getRegionName() {
        //connect with master and get path name
        try {
            //create a socket with master server, tell region's ip
            Socket masterSocket = new Socket(MASTER_SERVER_IP, MASTER_PORT);
            OutputStream outputStream = masterSocket.getOutputStream();
            DataOutputStream out = new DataOutputStream(outputStream);
            out.writeUTF(REGION_SERVER_IP);
            InputStream inputStream = masterSocket.getInputStream();
            DataInputStream in = new DataInputStream(inputStream);
            String name = in.readUTF();
            masterSocket.close();
//            out.close();
//            in.close();
//            outputStream.close();
//            inputStream.close();
            return name;
        } catch (IOException e) {
            log.warn("socket connect with master is wrong");
            e.printStackTrace();
            return "error";
        }
    }

    public static void tableChange() {
        StringBuilder data = new StringBuilder(REGION_SERVER_IP);
        table_list = CatalogManager.get_tables();
        System.out.println("(tableChange) table_list:" + table_list);
        for (String t : table_list) {
            data.append(" ");
            data.append(t);
        }
        try {
            client.setData().forPath(path, data.toString().getBytes());
        } catch (Exception e) {
            log.warn("zk setData failed!");
            e.printStackTrace();
        }
    }

    public static InetAddress getLocalHostLANAddress() throws UnknownHostException {
        try {
            InetAddress candidateAddress = null;
            for (Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces(); networkInterfaces.hasMoreElements(); ) {
                NetworkInterface networkInterface = (NetworkInterface) networkInterfaces.nextElement();
                for (Enumeration<InetAddress> inetAddresses = networkInterface.getInetAddresses(); inetAddresses.hasMoreElements(); ) {
                    InetAddress inetAddress = (InetAddress) inetAddresses.nextElement();
                    if (!inetAddress.isLoopbackAddress()) {
                        if (inetAddress.isSiteLocalAddress()) {
                            return inetAddress;
                        } else if (candidateAddress == null) {
                            candidateAddress = inetAddress;
                        }
                    }
                }
            }
            if (candidateAddress != null) {
                return candidateAddress;
            }
            InetAddress jdkSuppliedAddress = InetAddress.getLocalHost();
            if (jdkSuppliedAddress == null) {
                throw new UnknownHostException("The JDK InetAddress.getLocalHost() method unexpectedly returned null.");
            }
            return jdkSuppliedAddress;
        } catch (Exception e) {
            UnknownHostException unknownHostException = new UnknownHostException(
                    "Failed to determine LAN address: " + e);
            unknownHostException.initCause(e);
            throw unknownHostException;
        }
    }

    public static void closeConnect() {
        client.close();
    }

    public static String getPath() {
        return path;
    }
}
